package com.oracle.calllog.etl.util;

import java.io.IOException;
import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.TreeSet;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.NamespaceDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

/******************************************
 * @ClassName HBaseUtil
 * @Description HBaseAPI操作工具类的简单封装
 * @Author Mr.Chen
 * @Date 2022/6/11 22:57
 * @Version 1.0
 *****************************************/
public class HBaseUtil {
    /**
     * 判断命名空间是否存在
     * @param conf hadoop配置对象
     * @param namespace 命名空间的名称
     * @return
     */
    public static boolean isExistNamespace(Configuration conf, String namespace) throws IOException {
        Connection connection = HBaseConnectionUtil.getConnection(conf);
        Admin admin = connection.getAdmin();

        NamespaceDescriptor[] listNamespaceDescriptors = admin.listNamespaceDescriptors();

        boolean exists=false;
        for (NamespaceDescriptor namespaceDescriptor : listNamespaceDescriptors) {
            if(namespaceDescriptor.getName().equals(namespace)) {
                exists=true;
                break;
            }
        }

        admin.close();

        return exists;
    }

    /**
     * 初始化命名空间
     * @param conf hadoop配置对象
     * @param namespace 命名空间的名称
     */
    public static void initNamespace(Configuration conf, String namespace) throws IOException {
        Connection connection = HBaseConnectionUtil.getConnection(conf);
        Admin admin = connection.getAdmin();
        NamespaceDescriptor nd = NamespaceDescriptor.create(namespace)
                .addConfiguration("CREATE_TIME", String.valueOf(System.currentTimeMillis()))
                .build();
        admin.createNamespace(nd);

        admin.close();
    }

    /**
     * 判断表是否存在
     *
     * @param conf hadoop配置对象
     * @param tableName 表名
     * @return
     */
    public static boolean isExistTable(Configuration conf, String tableName) throws IOException {
        Connection connection = HBaseConnectionUtil.getConnection(conf);
        Admin admin = connection.getAdmin();
        boolean exists = admin.tableExists(TableName.valueOf(tableName));

        admin.close();

        return exists;
    }


    /**
     * 创建表
     * @param conf hadoop配置对象
     * @param tableName 表名称
     * @param regionCount 业务的预分区数量（实际分区会比业务预分区多一个，用于收集异常数据）
     * @param columnFamily
     * @throws IOException
     */
    public static void createTable(Configuration conf, String tableName, int regionCount, String...columnFamily)
            throws IOException {
        Connection connection = HBaseConnectionUtil.getConnection(conf);
        Admin admin = connection.getAdmin();

        // 创建表格描述对象
        TableDescriptorBuilder tableDescBuilder = TableDescriptorBuilder.newBuilder(TableName.valueOf(tableName));
        // 构造列族描述对象并存入List集合
        List<ColumnFamilyDescriptor> familyDescriptors = new ArrayList<>();
        for (String cf : columnFamily) {
            // 创建列族描述对象
            ColumnFamilyDescriptor familyDescriptor = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(cf)).build();
            familyDescriptors.add(familyDescriptor);
        }
        tableDescBuilder.setColumnFamilies(familyDescriptors);

        admin.createTable(tableDescBuilder.build(), genSplitKeys(regionCount));
        admin.close();
    }

    /**
     * 用于生成预分区的RowKey分界点，而且要保证有序
     * @param regionCount 业务的预分区数量（实际分区会比业务预分区多一个，用于收集异常数据）
     * @return 预分区的分区临界点数组
     */
    private static byte[][] genSplitKeys(int regionCount) {
        // 定义一个存放分区键的数组
        String[] keys = new String[regionCount];
        // 目前推算，region个数不会超过2位数，所以region分区键格式化为两位数字所代表的字符串
        DecimalFormat df = new DecimalFormat("00");
        for (int i = 0; i < regionCount; i++) {
            keys[i] = df.format(i) + "|";
        }

        byte[][] splitKeys = new byte[regionCount][];
        // 生成byte[][]类型的分区键的时候，一定要保证分区键是有序的
        TreeSet<byte[]> treeSet = new TreeSet<>(Bytes.BYTES_COMPARATOR);
        for (int i = 0; i < regionCount; i++) {
            treeSet.add(Bytes.toBytes(keys[i]));
        }

        int index=0;
        for (Iterator<byte[]> it = treeSet.iterator();it.hasNext();) {
            byte[] b = it.next();
            splitKeys[index++] = b;
        }
        return splitKeys;
    }

    /**
     * 根据重要统计信息生成rowkey<BR>
     * 格式为：regionCode_contact1No_buildTime_contact2No_duration_flag
     * @return RowKey
     */
    public static String genRowKey(String regionCode, String contact1No, String buildTime,
                                   String contact2No, String duration, String flag) {
        StringBuilder builder = new StringBuilder();
        builder.append(regionCode + "_")
                .append(contact1No + "_")
                .append(buildTime + "_")
                .append(contact2No + "_")
                .append(duration + "_")
                .append(flag);
        return builder.toString();
    }

    /**
     * 根据手机号码和年月信息计算出rowkey的分区编号(RegionCode)
     * @param contact1No 第1位置的联系人手机号码
     * @param buildTime 通话时间
     * @param regions 分区数
     * @return 经过离散计算的rowkey所在的分区号
     */
    public static String genRegionCode(String contact1No, String buildTime, int regions) {
        // 取出年月
        String ym = buildTime.replaceAll("[\\D]+", "").substring(0, 6);
        // 离散操作:手机号码与年月拼接后的字符串取HashCode,针对有可能产生负数的结果(计算HashCode时溢出)强制转为正数
        int hashResult = (contact1No + ym).hashCode() & Integer.MAX_VALUE;
        // 生成分区号
        int regionCode = hashResult % regions;
        // 格式化分区号
        DecimalFormat df = new DecimalFormat("00");
        return df.format(regionCode);
    }
}
